package org.dromara.common.redis.utils;

import lombok.AccessLevel;
import lombok.NoArgsConstructor;
import org.dromara.common.core.utils.SpringUtils;
import org.redisson.api.*;

import java.util.concurrent.CompletionStage;
import java.util.concurrent.TimeUnit;
import java.util.function.Function;


/**
 * @Author 刘武贵
 * @Date 2024/8/15 22:00
 * @Description 功能描述：分布式队列工具
 * 轻量级队列 重量级数据量 请使用 MQ
 * 要求 redis 5.X 以上
 */
@NoArgsConstructor(access = AccessLevel.PRIVATE)
public class QueueUtils {

    /**
     * 定义一个静态常量CLIENT，用于获取RedissonClient实例
     * 通过SpringUtils工具类从Spring上下文中取出RedissonClient实例
     * 这里使用静态常量是因为RedissonClient实例在应用中会被频繁使用，且实例应该是全局唯一的
     * 选择在类加载时就获取实例，是为了提高应用的初始化效率，避免在每次使用时都去创建新的实例
     */
    private static final RedissonClient CLIENT = SpringUtils.getBean(RedissonClient.class);


    /**
     * 获取客户端实例
     */
    public static RedissonClient getClient() {
        return CLIENT;
    }

    /**
     * 添加普通队列数据
     * 向指定的阻塞队列中添加元素
     *
     * 此方法通过队列名称从Redis客户端获取一个阻塞队列，并将数据元素添加到该队列中
     * 如果队列已存在，则将数据添加到队列的末尾；如果队列不存在，则创建一个新的队列并添加数据
     *
     * @param queueName 队列名
     * @param data 要添加到队列中的数据元素，类型泛型
     * @param <T> 泛型参数，表示队列中元素的类型
     * @return 如果成功将数据添加到队列中，则返回true；如果添加失败，则返回false
     */
    public static <T> boolean addQueueObject(String queueName, T data) {
        // 通过队列名称获取Redis中的阻塞队列
        RBlockingQueue<T> queue = CLIENT.getBlockingQueue(queueName);
        // 将数据元素添加到队列中，如果队列不存在，则此操作将失败
        return queue.offer(data);
    }

    /**
     * 通用获取一个队列数据 没有数据返回 null(不支持延迟队列)
     * 从指定的阻塞队列中获取并移除一个对象
     *
     * @param queueName 队列名
     * @return 队列中的下一个元素，如果在给定时间内队列为空则返回null
     */
    public static <T> T getQueueObject(String queueName) {
        // 通过客户端获取指定名称的阻塞队列
        RBlockingQueue<T> queue = CLIENT.getBlockingQueue(queueName);
        // 从队列中获取并移除一个对象，如果队列为空则返回null
        return queue.poll();
    }

    /**
     * 通用删除队列数据(不支持延迟队列)
     * 从指定的队列中移除一个元素
     *
     * @param queueName 队列的名称，用于标识特定的队列
     * @param data 要从队列中移除的元素
     * @return 如果队列中存在并成功移除了指定元素，返回true；否则返回false
     */
    public static <T> boolean removeQueueObject(String queueName, T data) {
        // 根据队列名称获取对应的阻塞队列
        RBlockingQueue<T> queue = CLIENT.getBlockingQueue(queueName);
        return queue.remove(data);
    }

    /**
     * 通用销毁队列 所有阻塞监听 报错(不支持延迟队列)
     *
     * @param queueName 队列名称
     * @param <T>       队列中元素的类型
     * @return 如果队列销毁成功，返回true；否则返回false
     */
    public static <T> boolean destroyQueue(String queueName) {
        // 通过队列名称获取对应的阻塞队列
        RBlockingQueue<T> queue = CLIENT.getBlockingQueue(queueName);
        return queue.delete();
    }

    /**
     * 添加延迟队列数据 默认毫秒
     *
     * @param queueName 队列名，队列名称，用于标识具体的队列
     * @param data      数据，要添加到队列中的数据对象
     * @param time      延迟时间，对象在队列中的延迟时间，以毫秒为单位
     */
    public static <T> void addDelayedQueueObject(String queueName, T data, long time) {
        addDelayedQueueObject(queueName, data, time, TimeUnit.MILLISECONDS);
    }

    /**
     * 添加延迟队列数据
     * 向指定的延迟队列中添加对象
     * 该方法首先通过队列名称获取对应的阻塞队列，然后使用该阻塞队列获取延迟队列对象，
     * 最后将数据对象添加到延迟队列中，指定延迟时间和时间单位
     *
     * @param queueName 队列名，用于标识队列
     * @param data      数据 要添加到队列中的数据对象
     * @param time      延迟时间，表示数据对象需要延迟多长时间才能被处理
     * @param timeUnit  单位，表示上述延迟时间的时间单位
     */
    public static <T> void addDelayedQueueObject(String queueName, T data, long time, TimeUnit timeUnit) {
        // 根据队列名称获取阻塞队列
        RBlockingQueue<T> queue = CLIENT.getBlockingQueue(queueName);
        // 使用阻塞队列获取延迟队列对象
        RDelayedQueue<T> delayedQueue = CLIENT.getDelayedQueue(queue);
        // 向延迟队列中添加数据对象，指定延迟时间和时间单位
        delayedQueue.offer(data, time, timeUnit);
    }

    /**
     * 获取一个延迟队列数据 没有数据返回 null
     * 从延迟队列中获取对象
     *
     * 此方法首先根据队列名称获取一个阻塞队列，然后使用该阻塞队列获取对应的延迟队列对象，
     * 最后从延迟队列中取出一个到期了的对象返回
     *
     * @param queueName 队列名称，用于获取对应的阻塞队列和延迟队列
     * @param <T>       泛型参数，表示队列中对象的类型
     * @return 返回从延迟队列中取出的对象，如果队列为空或没有到期的对象，则返回null
     */
    public static <T> T getDelayedQueueObject(String queueName) {
        // 根据队列名称获取阻塞队列
        RBlockingQueue<T> queue = CLIENT.getBlockingQueue(queueName);
        // 使用阻塞队列获取对应的延迟队列
        RDelayedQueue<T> delayedQueue = CLIENT.getDelayedQueue(queue);
        // 从延迟队列中取出一个到期了的对象返回
        return delayedQueue.poll();
    }

    /**
     * 删除延迟队列数据
     * 从延迟队列中移除指定对象
     *
     * 此方法通过队列名称获取关联的延迟队列对象，并尝试从该延迟队列中移除指定的数据对象
     * 主要用于在Redisson提供的延迟队列机制中，手动移除不再需要延迟处理的对象，以提高队列管理的灵活性和效率
     *
     * @param queueName 延迟队列的名称，用于标识具体的延迟队列
     * @param data 需要从延迟队列中移除的具体数据对象
     * @return 返回移除操作的结果，true表示成功移除，false表示移除失败，可能的原因包括但不限于数据对象不在队列中
     */
    public static <T> boolean removeDelayedQueueObject(String queueName, T data) {
        // 通过队列名称获取关联的阻塞队列对象
        RBlockingQueue<T> queue = CLIENT.getBlockingQueue(queueName);
        // 通过阻塞队列对象获取关联的延迟队列对象
        RDelayedQueue<T> delayedQueue = CLIENT.getDelayedQueue(queue);
        // 尝试从延迟队列中移除指定的数据对象并返回操作结果
        return delayedQueue.remove(data);
    }

    /**
     * 销毁延迟队列 所有阻塞监听 报错
     *
     * @param queueName 队列名称
     */
    public static <T> void destroyDelayedQueue(String queueName) {
        // 获取指定名称的阻塞队列
        RBlockingQueue<T> queue = CLIENT.getBlockingQueue(queueName);
        // 获取与阻塞队列关联的延迟队列
        RDelayedQueue<T> delayedQueue = CLIENT.getDelayedQueue(queue);
        // 销毁延迟队列
        delayedQueue.destroy();
    }

    /**
     * 添加优先队列数据
     * 向指定的优先级队列中添加元素
     *
     * @param queueName 队列名，用于标识具体的优先级队列
     * @param data      数据 要添加到优先级队列中的元素
     * @param <T>       优先级队列中元素的类型
     * @return 添加成功返回true，否则返回false
     */
    public static <T> boolean addPriorityQueueObject(String queueName, T data) {
        // 根据队列名称获取对应的优先级队列实例
        RPriorityBlockingQueue<T> priorityBlockingQueue = CLIENT.getPriorityBlockingQueue(queueName);
        // 尝试将元素添加到优先级队列中，并返回添加结果
        return priorityBlockingQueue.offer(data);
    }

    /**
     * 优先队列获取一个队列数据 没有数据返回 null(不支持延迟队列)
     * 从指定的优先级队列中获取对象
     *
     * 此方法通过队列名称获取一个优先级阻塞队列，并尝试从该队列中.poll()方法获取一个对象
     * 如果队列中当前没有可用的对象，poll()方法将返回null
     *
     * @param queueName 队列的名称，用于从客户端获取对应的优先级阻塞队列
     * @return 从优先级队列中获取的对象，或在队列为空时返回null
     */
    public static <T> T getPriorityQueueObject(String queueName) {
        // 使用泛型以支持不同类型的对象
        // 通过队列名称从客户端获取优先级阻塞队列
        RPriorityBlockingQueue<T> queue = CLIENT.getPriorityBlockingQueue(queueName);
        // 尝试从队列中获取一个对象，如果队列为空则返回null
        return queue.poll();
    }

    /**
     * 优先队列删除队列数据(不支持延迟队列)
     * 从指定的优先级队列中移除一个对象
     *
     * @param queueName 队列名称，用于标识具体的优先级队列
     * @param data 要从队列中移除的对象
     * @return 如果队列中存在并成功移除了指定对象，返回true；否则返回false
     */
    public static <T> boolean removePriorityQueueObject(String queueName, T data) {
        // 根据队列名称获取对应的优先级队列实例
        RPriorityBlockingQueue<T> queue = CLIENT.getPriorityBlockingQueue(queueName);
        return queue.remove(data);
    }

    /**
     * 优先队列销毁队列 所有阻塞监听 报错(不支持延迟队列)
     * 销毁指定的优先级队列
     *
     * @param queueName 队列名称
     * @param <T>       队列中元素的类型
     * @return 如果销毁成功，返回true；否则返回false
     */
    public static <T> boolean destroyPriorityQueue(String queueName) {
        // 根据队列名称获取优先级阻塞队列实例
        RPriorityBlockingQueue<T> queue = CLIENT.getPriorityBlockingQueue(queueName);
        return queue.delete();
    }

    /**
     * 尝试设置 有界队列 容量 用于限制数量
     * 此方法的作用是通过队列名称获取存在的有界队列对象，并尝试设置其容量
     * 它不关心队列中元素的类型，因为类型泛型<T>允许它操作任何类型的有界队列
     * 之所以是“尝试”设置，是因为队列的容量可能由于各种原因无法被更改到指定的值，
     * 比如队列当前的状态不允许更改容量或者指定的容量值无效
     *
     * @param queueName 队列名
     * @param capacity  容量
     * @return 返回设置是否成功的布尔值，如果设置成功返回true，否则返回false
     */
    public static <T> boolean trySetBoundedQueueCapacity(String queueName, int capacity) {
        // 根据给定的队列名称，尝试设置队列的容量。
        // 此操作旨在对队列的容量进行动态调整，以提高系统资源的利用效率或满足新的业务需求。
        RBoundedBlockingQueue<T> boundedBlockingQueue = CLIENT.getBoundedBlockingQueue(queueName);
        return boundedBlockingQueue.trySetCapacity(capacity);
    }

    /**
     * 尝试设置 有界队列 容量 用于限制数量
     * 尝试设置有界队列的容量
     * 如果队列存在且设置了在存在时销毁，则销毁队列
     *
     * @param queueName 队列名
     * @param capacity  容量
     * @param destroy   已存在是否销毁
     * @return 如果成功设置队列容量或按照要求销毁队列，则返回true；否则返回false
     */
    public static <T> boolean trySetBoundedQueueCapacity(String queueName, int capacity, boolean destroy) {
        // 获取指定名称的有界阻塞队列
        RBoundedBlockingQueue<T> boundedBlockingQueue = CLIENT.getBoundedBlockingQueue(queueName);
        // 检查队列是否存在以及是否设置了销毁标志
        if (boundedBlockingQueue.isExists() && destroy) {
            // 销毁已存在的队列
            destroyQueue(queueName);
        }
        // 尝试设置队列的容量并返回结果
        return boundedBlockingQueue.trySetCapacity(capacity);
    }

    /**
     * 添加有界队列数据
     *
     * @param queueName 队列名
     * @param data      数据
     * @return 添加成功 true 已达到界限 false
     */
    public static <T> boolean addBoundedQueueObject(String queueName, T data) {
        // 根据队列名称从客户端获取对应的有界阻塞队列
        RBoundedBlockingQueue<T> boundedBlockingQueue = CLIENT.getBoundedBlockingQueue(queueName);
        // 尝试将数据对象添加到有界阻塞队列中，如果成功添加则返回true，否则返回false
        return boundedBlockingQueue.offer(data);
    }

    /**
     * 有界队列获取一个队列数据 没有数据返回 null(不支持延迟队列)
     * 从指定的有界队列中获取并移除一个对象
     *
     * @param queueName 队列名称
     * @param <T>       队列中对象的类型
     * @return 从队列中获取的对象，如果队列为空返回null
     */
    public static <T> T getBoundedQueueObject(String queueName) {
        // 根据队列名称从客户端获取有界阻塞队列
        RBoundedBlockingQueue<T> queue = CLIENT.getBoundedBlockingQueue(queueName);
        // 从队列中移除并返回等待时间最久的元素，如果队列为空则返回null
        return queue.poll();
    }

    /**
     * 有界队列删除队列数据(不支持延迟队列)
     * 从指定的有界队列中移除指定的对象
     *
     * @param queueName 队列名称
     * @param data      要移除的对象
     * @param <T>       队列中对象的类型
     * @return 如果队列中存在并移除了指定对象，返回true；否则返回false
     */
    public static <T> boolean removeBoundedQueueObject(String queueName, T data) {
        // 根据队列名称获取具体的有界阻塞队列实例
        RBoundedBlockingQueue<T> queue = CLIENT.getBoundedBlockingQueue(queueName);
        return queue.remove(data);
    }

    /**
     * 有界队列销毁队列 所有阻塞监听 报错(不支持延迟队列)
     * 销毁指定的有界队列，同时删除所有阻塞监听
     *
     * @param queueName 队列名称
     * @param <T>       队列中对象的类型
     * @return 如果队列被成功销毁，返回true；否则返回false
     */
    public static <T> boolean destroyBoundedQueue(String queueName) {
        // 通过队列名称从客户端获取有界阻塞队列
        RBoundedBlockingQueue<T> queue = CLIENT.getBoundedBlockingQueue(queueName);
        return queue.delete();
    }

    /**
     * 订阅阻塞队列(可订阅所有实现类 例如: 延迟 优先 有界 等)
     *
     * @param queueName 队列名称
     * @param consumer 消费者函数，接受一个泛型参数T，并返回一个CompletionStage<Void>
     * @param isDelayed 是否是延迟队列
     */
    public static <T> void subscribeBlockingQueue(String queueName, Function<T, CompletionStage<Void>> consumer, boolean isDelayed) {
        // 获取阻塞队列实例
        RBlockingQueue<T> queue = CLIENT.getBlockingQueue(queueName);
        // 如果是延迟队列，则订阅延迟队列
        if (isDelayed) {
            // 订阅延迟队列
            CLIENT.getDelayedQueue(queue);
        }
        // 订阅队列中的元素
        queue.subscribeOnElements(consumer);
    }

}
